package com.gmscosmo.janus.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.gmscosmo.janus.util.JanusLogUtil;
import com.gmscosmo.janus.util.JanusUtil;
import com.gmscosmo.model.Node;
import com.gmscosmo.model.Producer;
import com.gmscosmo.model.Room;
import com.gmscosmo.model.Subscriber;
import com.gmscosmo.janus.util.Utils;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * ClassName: DistributedIMJanusUtil
 * Description:
 * Date: 2021/10/11 18:37
 *
 * @author stephen.qiu
 * @version 1.0
 */
@Service
public class DistributedIMJanusUtil extends JanusUtil {
    private static final String PREFIX_MAX_PARTICIPANT = "MAX_PARTICIPANT";
    private static final String PREFIX_IS_SHARE_STREAM = "IS_SHARE_STREAM";
    private static final String PREFIX_LOAD_DISTRIBUTION_STRATEGY = "LOAD_DISTRIBUTION_STRATEGY";
    private static final String PREFIX_ROOM_NODEID = "ROOM_NODEID";
    private static final String PREFIX_SUBSCRIBERS = "SUBSCRIBERS";
    private static final String PREFIX_PRODUCERS = "PRODUCERS";
    private static final String PREFIX_USER_NODE = "USER_NODE";
    private static final String PREFIX_NODE_INFO = "NODE_INFO";
    private static final String PREFIX_USER_ROOM = "USER_ROOM";
    private static final String PREFIX_ROOM_INFO = "ROOM_INFO";
    private static final String PREFIX_PORT = "PORT";

    @PostConstruct
    public void init() {
        janusLogUtil.log("DistributedIMJanusUtil初始化开始",this.getClass(), JanusLogUtil.TRACE);
        ConcurrentHashMap<Integer, Node> nodeInfos = new ConcurrentHashMap<>(16);
        try {
            nodeInfos = getJanusServer();
        } catch (Exception e) {
            e.printStackTrace();
        }
        int MAX_PARTICIPANT = janusConfig.getMax_participant();
        int LOAD_DISTRIBUTION_STRATEGY = janusConfig.getLoad_distribution_strategy();
        boolean IS_SHARE_STREAM = janusConfig.isIs_share_stream();
        messagesStore.putObject(PREFIX_MAX_PARTICIPANT, "", MAX_PARTICIPANT);
        janusLogUtil.log("设置服务器最大容纳人数:"+MAX_PARTICIPANT,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.putObject(PREFIX_LOAD_DISTRIBUTION_STRATEGY, "", LOAD_DISTRIBUTION_STRATEGY);
        janusLogUtil.log("设置服务器负载均衡策略:"+LOAD_DISTRIBUTION_STRATEGY,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.putObject(PREFIX_IS_SHARE_STREAM, "", IS_SHARE_STREAM);
        janusLogUtil.log("设置服务器是否共享流策略:"+IS_SHARE_STREAM,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.putObject(PREFIX_NODE_INFO, "", nodeInfos);
        for(Node node:nodeInfos.values()) {
            janusLogUtil.log("设置janus服务器节点:" + node, this.getClass(), JanusLogUtil.TRACE);
        }

    }

    public ConcurrentHashMap<Integer, Node> getJanusServer() throws Exception {
        List<String> queues = rabbitmqHttpRequestUtil.getQueueName("to");
        if (queues == null) {
            janusLogUtil.log("未发现janus服务器",this.getClass(),JanusLogUtil.WARN);
            throw new Exception("未发现janus服务器");
        }
        ConcurrentHashMap<Integer, Node> nodeInfos = new ConcurrentHashMap<>(16);
        Node node;
        for (String queue : queues) {
            node = new Node();
            String host = rabbitmqHttpRequestUtil.getQueueConsumerHost(queue);
            if (host == null) {
                continue;
            }
            int nodeId = Math.abs(host.hashCode());
            node.setNodeId(nodeId);
            node.setClusterId(0);
            node.setHost(host);
            janusLogUtil.log("加载janus服务器:"+ host,this.getClass(),JanusLogUtil.INFO);
            node.setInterHost(queue.split("to-janus")[1]);
            node.setTo_janus(queue);
            node.setParticipant(0);
            putNodeInfo(nodeId, node);
            nodeInfos.put(nodeId, node);
        }
        return nodeInfos;
    }

    /**
     * 查询房间是否在某一个janus节点
     *
     * @param roomId
     * @return
     */
    @Override
    public boolean isRoomNodeIdContainsKey(String roomId) {
        boolean res =messagesStore.isObjectExist(PREFIX_ROOM_NODEID, roomId);
        janusLogUtil.log("查询房间:"+roomId+"是否存在优先节点:"+res,this.getClass(),JanusLogUtil.TRACE);
        return res;
    }

    /**
     * 获取房间所在janus节点node
     *
     * @param roomId
     * @return
     */
    @Override
    public int getRoomNodeId(String roomId) {
        String object = messagesStore.getObject(PREFIX_ROOM_NODEID, roomId);
        int nodeId =  object == null ? getNodeInfos().keys().nextElement() : JSONObject.parseObject(object, Integer.class);
        janusLogUtil.log("获取房间:"+roomId+"优先节点:"+nodeId,this.getClass(),JanusLogUtil.TRACE);
        return nodeId;
    }

    /**
     * 更新房间所在janus节点node
     *
     * @param roomId
     * @return
     */
    @Override
    public void putRoomNodeId(String roomId, int nodeId) {
        janusLogUtil.log("更新房间:"+roomId+"优先节点:"+nodeId,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.putObject(PREFIX_ROOM_NODEID, roomId, nodeId);
    }

    /**
     * 删除房间所在janus节点node信息
     *
     * @param roomId
     * @return
     */
    @Override
    public void removeRoomNodeId(String roomId) {
        janusLogUtil.log("删除房间:"+roomId+"优先节点",this.getClass(),JanusLogUtil.TRACE);
        messagesStore.removeObject(PREFIX_ROOM_NODEID, roomId);
    }

    @Override
    public boolean getIsShareStream() {
        String object = messagesStore.getObject(PREFIX_IS_SHARE_STREAM, "");
        return object == null ? false : JSONObject.parseObject(object, Boolean.class);
    }

    @Override
    public int getMaxParticipant() {
        String object = messagesStore.getObject(PREFIX_MAX_PARTICIPANT, "");
        return object == null ? 0 : JSONObject.parseObject(object, Integer.class);
    }

    @Override
    public int getLoadDistributionStrategy() {
        String object = messagesStore.getObject(PREFIX_LOAD_DISTRIBUTION_STRATEGY, "");
        return object == null ? 0 : JSONObject.parseObject(object, Integer.class);
    }

    /**
     * 查询用户是否存在消费
     *
     * @param userId
     * @return
     */
    @Override
    public Boolean isSubscribersContains(String userId) {
        boolean res = messagesStore.isObjectExist(PREFIX_SUBSCRIBERS, userId);
        janusLogUtil.log("查询用户:"+userId+"是否存在消费者:"+res,this.getClass(),JanusLogUtil.TRACE);
        return res;
    }

    /**
     * 获取用户消费者列表
     *
     * @param userId
     * @return
     */
    @Override
    public List<Subscriber> getSubscribers(String userId) {
        String object = messagesStore.getObject(PREFIX_SUBSCRIBERS, userId);
        List<Subscriber> res = object == null ? new ArrayList<>() : JSONObject.parseObject(object, new TypeReference<List<Subscriber>>() {
        });
        janusLogUtil.log("获取用户:"+userId+"消费者列表:"+res,this.getClass(),JanusLogUtil.TRACE);
        return res;
    }

    /**
     * 移除用户消费者
     *
     * @param userId
     */
    @Override
    public void removeSubscribers(String userId) {
        janusLogUtil.log("移除用户:"+userId+"消费者列表",this.getClass(),JanusLogUtil.TRACE);
        messagesStore.removeObject(PREFIX_SUBSCRIBERS, userId);
    }

    /**
     * 更新用户消费者列表
     *
     * @param userId
     * @param subscriberList
     */
    @Override
    public void putSubscribers(String userId, List<Subscriber> subscriberList) {
        janusLogUtil.log("更新用户:"+userId+"消费者列表:"+subscriberList,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.putObject(PREFIX_SUBSCRIBERS, userId, subscriberList);
    }

    /**
     * 查询用户是否存在生产
     *
     * @param userId
     * @return
     */
    @Override
    public Boolean isProducersContains(String userId) {
        boolean res = messagesStore.isObjectExist(PREFIX_PRODUCERS, userId);
        janusLogUtil.log("查询用户:"+userId+"是否存在生产者:"+res,this.getClass(),JanusLogUtil.TRACE);
        return res;
    }

    /**
     * 获取用户生产者列表
     *
     * @param userId
     * @return
     */
    @Override
    public List<Producer> getProducers(String userId) {
        String object = messagesStore.getObject(PREFIX_PRODUCERS, userId);
        List<Producer> res =  object == null ? new ArrayList<>() : JSONObject.parseObject(object, new TypeReference<List<Producer>>() {
        });
        janusLogUtil.log("获取用户:"+userId+"生产者列表:"+res,this.getClass(),JanusLogUtil.TRACE);
        return res;
    }

    /**
     * 移除用户生产者
     *
     * @param userId
     */
    @Override
    public void removeProducers(String userId) {
        janusLogUtil.log("移除用户:"+userId+"生产者列表",this.getClass(),JanusLogUtil.TRACE);
        messagesStore.removeObject(PREFIX_PRODUCERS, userId);
    }

    /**
     * 更新用户生产者
     *
     * @param userId
     * @param producerList
     */
    @Override
    public void putProducers(String userId, List<Producer> producerList) {
        janusLogUtil.log("更新用户:"+userId+"生产者列表:"+producerList,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.putObject(PREFIX_PRODUCERS, userId, producerList);
    }

    /**
     * 查询用户是否任在某一个节点
     *
     * @param userId
     * @return
     */
    @Override
    public Boolean isUserNodeContains(String userId) {
        boolean res = messagesStore.isObjectExist(PREFIX_USER_NODE, userId);
        janusLogUtil.log("查询用户:"+userId+"是否任在某一个节点:"+res,this.getClass(),JanusLogUtil.TRACE);
        return res;
    }

    /**
     * 把用户从节点移除
     *
     * @param userId
     */
    @Override
    public void removeUserNode(String userId) {
        janusLogUtil.log("把用户:"+userId+"从节点移除",this.getClass(),JanusLogUtil.TRACE);
        messagesStore.removeObject(PREFIX_USER_NODE, userId);
    }

    /**
     * 获取用户当前所在节点
     *
     * @param userId
     * @return
     */
    @Override
    public int getUserNode(String userId) {
        String object = messagesStore.getObject(PREFIX_USER_NODE, userId);
        int nodeId =  object == null ? getNodeInfos().keys().nextElement() : JSONObject.parseObject(object, Integer.class);
        if(getNodeInfo(nodeId)==null){
            nodeId =getNodeInfos().keys().nextElement();
        }
        janusLogUtil.log("获取用户:"+userId+"当前所在节点:"+nodeId,this.getClass(),JanusLogUtil.TRACE);
        return nodeId;
    }

    /**
     * 添加用户到某节点
     *
     * @param userId
     */
    @Override
    public void putUserNode(String userId, int nodeId) {
        janusLogUtil.log("添加用户:"+userId+"到节点:"+nodeId,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.putObject(PREFIX_USER_NODE, userId, nodeId);
    }

    /**
     * 获取所有节点信息
     *
     * @return
     */
    @Override
    public ConcurrentHashMap<Integer, Node> getNodeInfos() {
        String object = messagesStore.getObject(PREFIX_NODE_INFO, "");
        return object == null ? new ConcurrentHashMap<>() : JSON.parseObject(object, new TypeReference<ConcurrentHashMap<Integer, Node>>() {
        });
    }

    /**
     * 获取janus节点信息
     *
     * @param nodeId
     * @return
     */
    @Override
    public Node getNodeInfo(int nodeId) {
        return getNodeInfos().get(nodeId);
    }

    /**
     * 添加janus节点信息
     *
     * @param nodeId
     * @param node
     * @return
     */
    @Override
    public boolean putNodeInfo(int nodeId, Node node) {
        ConcurrentHashMap<Integer, Node> nodeInfos = getNodeInfos();
        nodeInfos.put(nodeId, node);
        janusLogUtil.log("添加janus:"+nodeId+"节点信息:"+node,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.putObject(PREFIX_NODE_INFO, "", nodeInfos);
        return true;
    }

    /**
     * 更新janus节点信息
     *
     * @param nodeId
     * @param node
     */
    @Override
    public boolean updateNodeInfo(int nodeId, Node node) {
        ConcurrentHashMap<Integer, Node> nodeInfos = getNodeInfos();
        boolean res = false;
        if (!nodeInfos.containsKey(nodeId)) {
            return false;
        }
        Node oldNode = nodeInfos.get(nodeId);
        if (oldNode.getParticipant() != 0) {
            return false;
        }
        if (!Utils.isEmpty(node.getClusterId())) {
            oldNode.setClusterId(node.getClusterId());
            res = true;
        }
        if (!Utils.isEmpty(node.getHost())) {
            oldNode.setHost(node.getHost());
            res = true;
        }
        if (!Utils.isEmpty(node.getInterHost())) {
            oldNode.setInterHost(node.getInterHost());
            res = true;
        }
        if (!Utils.isEmpty(node.getTo_janus())) {
            oldNode.setTo_janus(node.getTo_janus());
            res = true;
        }
        if (res) {
            nodeInfos.put(nodeId, oldNode);
            janusLogUtil.log("更新janus:"+nodeId+"节点信息:"+node,this.getClass(),JanusLogUtil.TRACE);
            messagesStore.putObject(PREFIX_NODE_INFO, "", nodeInfos);
        }
        return res;
    }

    /**
     * 移除janus节点信息
     *
     * @param nodeId
     */
    @Override
    public boolean removeNodeInfo(int nodeId) {
        ConcurrentHashMap<Integer, Node> nodeInfos = getNodeInfos();
        if (!nodeInfos.containsKey(nodeId)) {
            return false;
        }
        if (nodeInfos.get(nodeId).getParticipant() != 0) {
            return false;
        }
        nodeInfos.remove(nodeId);
        messagesStore.putObject(PREFIX_NODE_INFO, "", nodeInfos);
        janusLogUtil.log("移除janus:"+nodeId+"节点信息",this.getClass(),JanusLogUtil.TRACE);
        return true;
    }

    /**
     * 查询用户是否再某一个房间
     *
     * @param userId
     * @return
     */
    @Override
    public boolean isUserRoomContainsKey(String userId) {
        boolean res = messagesStore.isObjectExist(PREFIX_USER_ROOM, userId);
        janusLogUtil.log("查询用户:"+userId+"是否在某一个房间:"+res,this.getClass(),JanusLogUtil.TRACE);
        return res;
    }

    /**
     * 查询用户房间号
     *
     * @param userId
     * @return
     */
    @Override
    public String getUserRoom(String userId) {
        String object = messagesStore.getObject(PREFIX_USER_ROOM, userId);
        String roomId =  object == null ? null : JSONObject.parseObject(object, String.class);
        janusLogUtil.log("查询用户:"+userId+"房间号:"+roomId,this.getClass(),JanusLogUtil.TRACE);
        return roomId;
    }

    /**
     * 更新用户房间号
     *
     * @param userId
     * @param roomId
     */
    @Override
    public void putUserRoom(String userId, String roomId) {
        janusLogUtil.log("更新用户:"+userId+"房间号:"+roomId,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.putObject(PREFIX_USER_ROOM, userId, roomId);
    }

    /**
     * 移除用户房间号
     *
     * @param userId
     */
    @Override
    public void removeUserRoom(String userId) {
        janusLogUtil.log("移除用户:"+userId+"房间号",this.getClass(),JanusLogUtil.TRACE);
        messagesStore.removeObject(PREFIX_USER_ROOM, userId);
    }

    /**
     * 查询房间是否存在
     *
     * @param roomId
     * @return
     */
    @Override
    public boolean isRoomContainsKey(String roomId) {
        boolean res = messagesStore.isObjectExist(PREFIX_ROOM_INFO, roomId);
        janusLogUtil.log("查询房间:"+roomId+"是否存在",this.getClass(),JanusLogUtil.TRACE);
        return res;
    }

    /**
     * 使用roomId 获取房间信息
     *
     * @param roomId
     * @return
     */
    @Override
    public Room getRoom(String roomId) {
        String object = messagesStore.getObject(PREFIX_ROOM_INFO, roomId);
        Room room = object == null ? null : JSONObject.parseObject(object, Room.class);
        janusLogUtil.log("查询房间:"+roomId+"信息:"+room,this.getClass(),JanusLogUtil.TRACE);
        return room;
    }

    /**
     * 更新房间信息
     *
     * @param roomId
     * @param room
     */
    @Override
    public void putRoom(String roomId, Room room) {
        janusLogUtil.log("更新房间:"+roomId+"信息:"+room,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.putObject(PREFIX_ROOM_INFO, roomId, room);
    }

    /**
     * 移除房间
     *
     * @param roomId
     */
    @Override
    public void removeRoom(String roomId) {
        janusLogUtil.log("移除房间:"+roomId,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.removeObject(PREFIX_ROOM_INFO, roomId);
    }

    /**
     * 释放端口
     *
     * @param port
     */
    @Override
    public void removePort(int port) {
        janusLogUtil.log("释放端口:"+port,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.removeObject(PREFIX_PORT, String.valueOf(port));
    }

    /**
     * 查询端口是否被占用
     *
     * @param port
     */
    @Override
    public boolean isPortContains(int port) {
        boolean res = messagesStore.isObjectExist(PREFIX_PORT, String.valueOf(port));
        janusLogUtil.log("查询端口:"+port+"是否被占用:"+res,this.getClass(),JanusLogUtil.TRACE);
        return messagesStore.isObjectExist(PREFIX_PORT, String.valueOf(port));
    }

    /**
     * 添加使用端口
     *
     * @param port
     */
    @Override
    public void addPort(int port) {
        janusLogUtil.log("添加使用端口:"+port,this.getClass(),JanusLogUtil.TRACE);
        messagesStore.putObject(PREFIX_PORT, String.valueOf(port), 1);
    }
}
